package com.atguigu.tingshu.search.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Operator;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.enums.search.SortFieldType;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumChannelVo;
import com.atguigu.tingshu.vo.search.AlbumDetailsVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.LocalDateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;

    @Autowired
    UserInfoFeignClient userInfoFeignClient;

    @Autowired
    ElasticsearchClient elasticsearchClient;

    @Autowired
    CategoryFeignClient categoryFeignClient;

    @Autowired
    ExecutorService executorService;

    @Autowired
    RedisTemplate redisTemplate;

    @Override
    public void upperAlbum(Long albumId) {
        Result<AlbumInfoIndex> result = albumInfoFeignClient.getAlbumInfoById(albumId);
        Assert.notNull(result, "该专辑不存在！");
        AlbumInfoIndex albumInfoIndex = result.getData();
        // 热度：根据上述指标结合一定的系数进行计算
        Integer playStatNum = albumInfoIndex.getPlayStatNum();
        Integer subscribeStatNum = albumInfoIndex.getSubscribeStatNum();
        Integer commentStatNum = albumInfoIndex.getCommentStatNum();
        Integer buyStatNum = albumInfoIndex.getBuyStatNum();
        double hotScore = playStatNum * 0.2 + subscribeStatNum * 0.3 + commentStatNum * 0.4 + buyStatNum * 0.1;
        albumInfoIndex.setHotScore(hotScore);
        elasticsearchTemplate.save(albumInfoIndex);
    }

    @Override
    public void downAlbum(Long albumId) {
        elasticsearchTemplate.delete(albumId.toString(), AlbumInfoIndex.class);
    }

    @Override
    public AlbumSearchResponseVo albumSearch(AlbumIndexQuery albumIndexQuery) {
        AlbumSearchResponseVo vo = new AlbumSearchResponseVo();
        vo.setPageNo(albumIndexQuery.getPageNo());
        Integer pageSize = albumIndexQuery.getPageSize();
        vo.setPageSize(pageSize);
        SearchRequest searchRequest = buildDsl(albumIndexQuery);
        try {
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
            // 解析查询结果，封装响应内容
            List<Hit<AlbumInfoIndex>> hits = response.hits().hits();
            if (CollUtil.isNotEmpty(hits)) {
                for (Hit<AlbumInfoIndex> hit : hits) {
                    AlbumInfoIndex source = hit.source();
                    AlbumInfoIndexVo albumInfoIndexVo = BeanUtil.copyProperties(source, AlbumInfoIndexVo.class);
                    // 高亮替换
                    List<String> list = hit.highlight().get("albumTitle");
                    if (CollUtil.isNotEmpty(list)) {
                        albumInfoIndexVo.setAlbumTitle(list.get(0));
                    }
                    vo.getList().add(albumInfoIndexVo);
                }
            }
            long total = response.hits().total().value();
            vo.setTotal(total);
            Long totalPage = total / pageSize;
            vo.setTotalPages(total % pageSize == 0 ? totalPage : totalPage + 1);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return vo;
    }

    @Override
    public AlbumDetailsVo getAlbumDetails(Long albumId) {
        AlbumDetailsVo vo = new AlbumDetailsVo();
        // 根据专辑id查询专辑信息
        CompletableFuture<AlbumInfo> albumInfoFuture = CompletableFuture.supplyAsync(() -> {
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            Assert.notNull(albumInfoResult, "专辑为空！");
            AlbumInfo albumInfo = albumInfoResult.getData();
            vo.setAlbumInfo(albumInfo);
            log.info("albumInfo:{}", JSON.toJSONString(albumInfo));
            return albumInfo;
        }, executorService);
        // 根据专辑id查询专辑统计信息
        CompletableFuture<Void> albumStatFuture = CompletableFuture.runAsync(() -> {
            Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStatByAlbumId(albumId);
            AlbumStatVo albumStatVo = albumStatVoResult.getData();
            vo.setAlbumStatVo(albumStatVo);
            log.info("albumStatVo:{}", JSON.toJSONString(albumStatVo));
        }, executorService);
        // 根据三级分类id查询分类信息
        CompletableFuture<Void> categoryFuture = albumInfoFuture.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> baseCategoryViewResult = categoryFeignClient.getBaseCategoryByC3Id(albumInfo.getCategory3Id());
            BaseCategoryView baseCategoryView = baseCategoryViewResult.getData();
            vo.setBaseCategoryView(baseCategoryView);
            log.info("baseCategoryView:{}", JSON.toJSONString(baseCategoryView));
        }, executorService);
        // 根据用户id查询主播信息
        CompletableFuture<Void> announcerFuture = albumInfoFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            UserInfoVo userInfoVo = userInfoVoResult.getData();
            vo.setAnnouncer(userInfoVo);
            log.info("announcer:{}", JSON.toJSONString(userInfoVo));
        }, executorService);
        CompletableFuture.allOf(albumStatFuture, categoryFuture, announcerFuture).join();
        return vo;
    }

    @Override
    public List<AlbumChannelVo> channel(Long category1Id) {
        // 根据一级分类id查询置顶的三级分类
        Result<List<BaseCategory3>> topBaseCategory3Result = this.categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(topBaseCategory3Result, "首页热门频道：获取置顶的三级分类失败！" + category1Id);
        List<BaseCategory3> baseCategory3s = topBaseCategory3Result.getData();
        Assert.notNull(baseCategory3s, "首页热门频道：没有获取到置顶的三级分类！" + category1Id);
        // 获取置顶的三级分类id集合
        List<FieldValue> category3Ids = baseCategory3s.stream().map(baseCategory3 -> FieldValue.of(baseCategory3.getId())).collect(Collectors.toList());
        // 为了方便根据三级分类id获取三级分类，把三级分类List集合转化成Map<category3Id, category>集合
        Map<Long, BaseCategory3> category3Map = baseCategory3s.stream().collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        // 组装搜索条件
        SearchRequest request = SearchRequest.of(s -> s.index("albuminfo")
                .size(0)
                .query(q -> q.terms(t -> t.field("category3Id").terms(ts -> ts.value(category3Ids))))
                .aggregations("category3IdAgg", a -> a.terms(t -> t.field("category3Id"))
                        //Elasticsearch的top_hits聚合用于获取每个分桶（bucket）中，与聚合查询匹配的文档的一些顶部文档，即获取前几条数据
                        .aggregations("topAgg", agg -> agg.topHits(t -> t
                                .size(6)
                                .sort(st -> st
                                        .field(f -> f
                                                .field("hotScore")
                                                .order(SortOrder.Desc))))))
        );
        log.info("查询置顶频道页的热门数据，DSL语句为：{}", request);
        // 执行搜索
        SearchResponse<AlbumInfoIndex> response = null;
        try {
            response = this.elasticsearchClient.search(request, AlbumInfoIndex.class);
        } catch (IOException e) {
            e.printStackTrace();
            throw new GuiguException(ResultCodeEnum.SERVICE_ERROR);
        }
        log.info("从ElasticSearch中查询置顶频道页的热门数据，响应结果为：{}", response);
        // 结果集封装
        Aggregate category3IdAgg = response.aggregations().get("category3IdAgg");
        if (ObjectUtil.isEmpty(category3IdAgg)) {
            return null;
        }
        List<AlbumChannelVo> resultList = category3IdAgg.lterms().buckets().array().stream().map(bucket -> {
            AlbumChannelVo channelVo = new AlbumChannelVo();
            channelVo.setBaseCategory3(category3Map.get(bucket.key()));
            List<Hit<JsonData>> topAgg = bucket.aggregations().get("topAgg").topHits().hits().hits();
            List<AlbumInfoIndex> albumInfoIndexList = topAgg.stream()
                    .map(hit -> hit.source().to(AlbumInfoIndex.class)).collect(Collectors.toList());
            channelVo.setList(albumInfoIndexList);
            return channelVo;
        }).collect(Collectors.toList());
        return resultList;
    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String field) {
        // 构建搜索条件
        SearchRequest searchRequest = SearchRequest.of(s -> s.index("albuminfo")
                .query(sq -> sq.term(st -> st.field("category1Id").value(category1Id)))
                .sort(ss -> ss.field(sf -> sf.field(field).order(SortOrder.Desc)))
                .size(20)
        );
        log.info("获取排行榜排名数据的DSL语句为：{}", searchRequest);
        try {
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
            List<Hit<AlbumInfoIndex>> hitList = response.hits().hits();
            if (CollUtil.isNotEmpty(hitList)) {
                List<AlbumInfoIndexVo> albumInfoIndexVos = hitList.stream().map(hit ->
                        BeanUtil.copyProperties(hit.source(), AlbumInfoIndexVo.class)).collect(Collectors.toList());
                return albumInfoIndexVos;
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new GuiguException(ResultCodeEnum.SERVICE_ERROR);
        }
        return null;
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        SearchRequest request;
        // 判断用户输入的是否是汉字
        if (keyword.matches(PinYinUtils.hanziRegex)) {
            request = SearchRequest.of(s -> s.index("suggestinfo")
                    .suggest(sg -> sg.suggesters("keywordSuggest",
                            ks -> ks.prefix(keyword).completion(c -> c.field("keyword").size(10).skipDuplicates(true)))));
        } else {
            request = SearchRequest.of(s -> s.index("suggestinfo")
                    .suggest(sg -> sg
                            .suggesters("keywordPinyinSuggest",
                                    ks -> ks.prefix(keyword).completion(c -> c.field("keywordPinyin").size(10).skipDuplicates(true)))
                            .suggesters("keywordSequence",
                                    ks -> ks.prefix(keyword).completion(c -> c.field("keywordSequence").size(10).skipDuplicates(true)))
                    )
            );
        }
        try {
            log.info(request.toString());
            SearchResponse<SuggestIndex> response = elasticsearchClient.search(request, SuggestIndex.class);
            log.info(response.toString());
            // 解析提示词结果集，获取提示词的title集合，并组装响应数据
            List<String> results = new ArrayList<>();
            Map<String, List<Suggestion<SuggestIndex>>> suggest = response.suggest();
            if (keyword.matches(PinYinUtils.hanziRegex)) {
                List<CompletionSuggestOption<SuggestIndex>> keywordOps = suggest.get("keywordSuggest")
                        .get(0).completion().options();
                List<String> keywordList = keywordOps.stream().map(ops -> ops.source().getTitle())
                        .collect(Collectors.toList());
                if (CollUtil.isNotEmpty(keywordList)) {
                    results.addAll(keywordList);
                }
            } else {
                List<CompletionSuggestOption<SuggestIndex>> keywordPinyinOps = suggest.get("keywordPinyinSuggest")
                        .get(0).completion().options();
                List<String> keywordPinyinList = keywordPinyinOps.stream().map(ops -> ops.source().getTitle())
                        .collect(Collectors.toList());
                List<CompletionSuggestOption<SuggestIndex>> keywordSequenceOps = suggest.get("keywordSequence")
                        .get(0).completion().options();
                List<String> keywordSequenceList = keywordSequenceOps.stream().map(ops -> ops.source().getTitle())
                        .collect(Collectors.toList());
                if (CollUtil.isNotEmpty(keywordPinyinList)) {
                    results.addAll(keywordPinyinList);
                }
                if (CollUtil.isNotEmpty(keywordSequenceList)) {
                    results.addAll(keywordSequenceList);
                }
            }
            // 大集合中的数据可能会存在重复的数据
            Set<String> set = new HashSet<>(); // 无序不重复
            set.addAll(results);
            results.clear();
            results.addAll(set);
            // 判断记录数是否太少，如果太少还可以联想出相关的数据进行提示
            if (results.size() >= 20) {
                return results;
            }
            return results;
        } catch (IOException e) {
            e.printStackTrace();
            throw new GuiguException(ResultCodeEnum.SERVICE_ERROR);
        }
    }

    @Override
    public void updateLatelyAlbumStat() {
        log.info("同步专辑统计信息开始：{}", System.currentTimeMillis());
        try {
            // redis会记录上一次执行数据同步时的截止时间，本次数据同步的范围为：上一次的截止时间 - 当前时间
            String startTime = (String) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_STAT_ENDTIME);
            if (StringUtils.isBlank(startTime)){
                // 如果没有则取一个小时前的时间
                startTime = LocalDateTime.now().minusHours(1).toString("yyyy-MM-dd HH:mm:ss");
            }
            // 统计截止时间
            String endTime = LocalDateTime.now().toString("yyyy-MM-dd HH:mm:ss");
            // 记录到redis作为下次统计的起始时间
            this.redisTemplate.opsForValue().set(RedisConstant.ALBUM_STAT_ENDTIME, endTime);
            // 获取统计的时间范围内，信息发生变化的专辑id集合
            Result<List<Long>> albumIdResult = this.albumInfoFeignClient.findLatelyUpdateAlbum(startTime, endTime);
            Assert.notNull(albumIdResult, "同步数据时，获取专辑列表失败");
            List<Long> albumIds = albumIdResult.getData();
            if (CollectionUtils.isEmpty(albumIds)) return;
            // 为了提高性能，每1000个专辑id拆分成一个部分，然后使用多线程并发执行
            List<List<Long>> partitions = Lists.partition(albumIds, 1000);
            int count = partitions.size();
            // 初始化countDownLatch
            CountDownLatch countDownLatch = new CountDownLatch(count);
            partitions.forEach(partAlbumIds -> {
                // 通过线程池控制线程数
                executorService.execute(() -> {
                    // 查询这部分专辑的统计信息
                    Result<List<AlbumStatVo>> albumStatResult = this.albumInfoFeignClient.findAlbumStatVoList(partAlbumIds);
                    Assert.notNull(albumStatResult, "同步数据到es时，获取专辑统计信息失败！");
                    List<AlbumStatVo> albumStatVos = albumStatResult.getData();
                    if (CollectionUtils.isEmpty(albumStatVos)) return;
                    // 遍历同步到es
                    albumStatVos.forEach(albumStatVo -> {
                        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
                        BeanUtils.copyProperties(albumStatVo, albumInfoIndex);
                        albumInfoIndex.setId(albumStatVo.getAlbumId());
                        // 设置热度
                        double hotScore = 0.2 * albumStatVo.getPlayStatNum()
                                + 0.3 * albumStatVo.getSubscribeStatNum()
                                + 0.4 * albumStatVo.getBuyStatNum()
                                + 0.1 * albumStatVo.getCommentStatNum();
                        albumInfoIndex.setHotScore(hotScore);
                        this.elasticsearchTemplate.update(albumInfoIndex);
                    });
                });
                countDownLatch.countDown();
            });
            countDownLatch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            log.info("同步专辑统计信息结束：{}", System.currentTimeMillis());
        }
    }

    /**
     * 构建首页检索的DSL语句
     *
     * @param query
     * @return
     */
    private SearchRequest buildDsl(AlbumIndexQuery query) {
        SearchRequest.Builder requestBuilder = new SearchRequest.Builder();
        requestBuilder.index("albuminfo");
        Query.Builder queryBuilder = new Query.Builder();
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        String keyword = query.getKeyword();
        if (StringUtils.isNotBlank(keyword)) {
            // 关键字匹配查询
            boolQueryBuilder.must(m -> m.multiMatch(mm -> mm
                    .query(keyword)
                    .fields("albumTitle", "albumIntro", "announcerName")
                    .operator(Operator.And)));
            // 高亮
            requestBuilder.highlight(h -> h
                    .fields("albumTitle", f -> f
                            .preTags("<font style='color:red'>").postTags("</font>")));
        }
        // 构建过滤条件
        Long category1Id = query.getCategory1Id();
        if (category1Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        Long category2Id = query.getCategory2Id();
        if (category2Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));
        }
        Long category3Id = query.getCategory3Id();
        if (category3Id != null) {
            boolQueryBuilder.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
        }
        List<String> attributeList = query.getAttributeList();
        if (!CollectionUtils.isEmpty(attributeList)) {
            attributeList.forEach(attribute -> {
                if (StringUtils.isBlank(attribute)) {
                    return;
                }
                String[] attrs = StringUtils.split(attribute, ":");
                if (attrs == null || attrs.length != 2 || !StringUtils.isNumeric(attrs[0]) || !StringUtils.isNumeric(attrs[1])) {
                    return;
                }
                // 嵌套过滤
                boolQueryBuilder.filter(f -> f.nested(n -> n
                        .path("attributeValueIndexList")
                        .query(q -> q.bool(b -> b
                                .must(m -> m.term(t -> t.field("attributeValueIndexList.attributeId").value(attrs[0])))
                                .must(m -> m.term(t -> t.field("attributeValueIndexList.valueId").value(attrs[1])))))));
            });
        }
        requestBuilder.query(queryBuilder.bool(boolQueryBuilder.build()).build());
        // 排序
        String order = query.getOrder();
        if (StringUtils.isNotBlank(order)) {
            String[] orders = StringUtils.split(order, ":");
            if (orders != null && orders.length == 2 && StringUtils.isNumeric(orders[0]) && Arrays.asList("desc", "asc").contains(orders[1])) {
                // 判断升降序
                SortOrder sortOrder = StringUtils.equals("desc", orders[1]) ? SortOrder.Desc : SortOrder.Asc;
                for (SortFieldType fieldType : SortFieldType.values()) {
                    if (fieldType.getCode().equals(orders[0])) {
                        requestBuilder.sort(rs -> rs.field(rf -> rf.field(fieldType.getField()).order(sortOrder)));
                    }
                }
            }
        }
        // 分页
        Integer pageSize = query.getPageSize();
        requestBuilder.from((query.getPageNo() - 1) * pageSize);
        requestBuilder.size(pageSize);
        // 结果集过滤
        requestBuilder.source(s -> s.filter(f -> f.excludes("hotScore")));
        SearchRequest request = requestBuilder.build();
        log.info("首页检索，DSL语句为：{}", request);
        return request;
    }
}
